package org.caw.hbase.explorer.filter;

import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.FilterList.Operator;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;
import org.caw.exception.CawParseException;
import org.caw.hbase.explorer.filter.model.ExpressionNode;
import org.caw.hbase.explorer.filter.model.Node;
import org.caw.hbase.explorer.filter.model.OperatorNode;

public class HBaseFilterParser {

	private Parser filterParser;

	public HBaseFilterParser() {
		setFilterParser(new Parser());
	}

	public Filter parse(String expression) throws CawParseException {
		Node node = filterParser.parseFilters(expression);
		return parse(node);
	}

	private Filter parse(Node node) throws CawParseException {

		if (node instanceof OperatorNode) {
			OperatorNode operator = (OperatorNode) node;
			FilterList filter = null;

			switch (operator.getLogicaOperator()) {
			case AND:
				filter = new FilterList(Operator.MUST_PASS_ALL);
				break;

			case OR:
				filter = new FilterList(Operator.MUST_PASS_ONE);
				break;
			}

			for (Node child : operator.getChilds()) {
				filter.addFilter(parse(child));
			}
			return filter;
		}
		if (node instanceof ExpressionNode) {
			ExpressionNode expressionNode = (ExpressionNode) node;

			byte[] value = null;
			if (expressionNode.getValue() == null) {
				value = new byte[0];
			} else {
				value = Bytes.toBytes(expressionNode.getValue());
			}

			SingleColumnValueFilter singleColumnValueFilter = null;

			switch (expressionNode.getOper()) {
			case EQ:
				singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes(expressionNode.getColumnFamily()), Bytes.toBytes(expressionNode.getQualifier()), CompareOp.EQUAL, value);
				break;
			case LT:
				singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes(expressionNode.getColumnFamily()), Bytes.toBytes(expressionNode.getQualifier()), CompareOp.LESS, value);
				break;
			case LTEQ:
				singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes(expressionNode.getColumnFamily()), Bytes.toBytes(expressionNode.getQualifier()), CompareOp.LESS_OR_EQUAL, value);
				break;
			case NEQ:
				singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes(expressionNode.getColumnFamily()), Bytes.toBytes(expressionNode.getQualifier()), CompareOp.NOT_EQUAL, value);
				break;
			case GT:
				singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes(expressionNode.getColumnFamily()), Bytes.toBytes(expressionNode.getQualifier()), CompareOp.GREATER, value);
				break;
			case GTEQ:
				singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes(expressionNode.getColumnFamily()), Bytes.toBytes(expressionNode.getQualifier()), CompareOp.GREATER_OR_EQUAL, value);
				break;
			case LIKE:
				singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes(expressionNode.getColumnFamily()), Bytes.toBytes(expressionNode.getQualifier()), CompareOp.EQUAL,
						new SubstringComparator(expressionNode.getValue()));
				break;
			case REGEX:
				singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes(expressionNode.getColumnFamily()), Bytes.toBytes(expressionNode.getQualifier()), CompareOp.EQUAL,
						new RegexStringComparator(expressionNode.getValue()));
				break;
			}

			singleColumnValueFilter.setFilterIfMissing(true);

			return singleColumnValueFilter;
		}
		return null;
	}

	public Parser getFilterParser() {
		return filterParser;
	}

	public void setFilterParser(Parser filterParser) {
		this.filterParser = filterParser;
	}

}
